package com.dmp

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext

/**
  * ClassName:AreaAnalyseRptV2
  * Package:com.dmp
  * Desciption:
  *
  * @date:2019 /8/23 19:17
  * @author:17611219021 @sina.cn
  */
object AreaAnalyseRptV2 {
  def main(args: Array[String]): Unit = {
    if(args.length !=2){
      println(
        """
          |com.dmp.fileToParquet
          |参数：
          | inputpath
          | outputPath
        """.stripMargin)
      sys.exit()
    }
    val conf: SparkConf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.setMaster("local[*]")
    //指定keyo序列化器与snappy格式，默认是gzip格式
    conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.sql.parquet.compression.codec","snappy")
    val sc: SparkContext = new SparkContext(conf)
    val Array(inputpath,outputPath)= args
    val sqlContext: SQLContext = new SQLContext(sc)

    val df1 = sqlContext.read.parquet(inputpath)

    val configuration = sc.hadoopConfiguration
    val fs = FileSystem.get(configuration)
    val path = new Path(outputPath)
    if(fs.exists(path)){
      fs.delete(path,true)
    }

    df1.map( row=>{
      val requestmode=row.getAs[String]("requestmode")
      val processnode = row.getAs[String]("processnode")
      val iseffective = row.getAs[String]("iseffective")
      val isbiding = row.getAs[String]("isbiding")
      val isbid = row.getAs[String]("isbid")
      val iswin = row.getAs[String]("iswin")
      val adorderid = row.getAs[String]("adorderid")
      val adcreativeid = row.getAs[String]("adcreativeid")
      val adpayment = row.getAs[String]("adpayment")
      val winprice = row.getAs[String]("winprice")

      //原始请求、有效请求、广告请求
      val listReq = RptUtils.calculateReq(requestmode,processnode)

      //参与竞价、竞价成功、消费、成本
      val listRtb = RptUtils.calculateRtb(iseffective,isbiding,isbid,adorderid,iswin,adpayment,winprice)

      //广告展示、广告点击
     val listShowOrClick = RptUtils.calculateShowOrClick(requestmode,iseffective)
      ((row.getAs[String]("province"),row.getAs[String]("area")),listReq++listRtb++listShowOrClick)

    }).reduceByKey((list1,list2) =>{
      list1.zip(list2).map(t=>{t._1+t._2})
    }).saveAsTextFile(outputPath)
    sc.stop()
  }

}
